package com.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.app.func.DimAsyncFunction;
import com.edu.realtime.bean.TradeCourseOrderBean;
import com.edu.realtime.util.DateFormatUtil;
import com.edu.realtime.util.MyClickhouseUtil;
import com.edu.realtime.util.MyKafkaUtil;
import com.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author zcx
 * @create 2022-10-20 11:35
 * 交易域：Course粒度下单业务过程聚合统计
 */
public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));

        /*// TODO 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 2.3 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        // 2.4 设置job取消后检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        // 2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        // 2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://nameservice1:8020/gmall/ck");
        // 2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME", "root");*/
        //TODO 3.从kafka的下单主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);



        //TODO 5.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //TODO 6.使用Flink的状态编程 + 定时器 实现去重
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastJsonObjState
                                = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class));
                    }


                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        //判断状态中是否保存了流中的数据
                        if (lastJsonObj == null) {
                            //如果状态中数据为null，说明当前处理的数据是该订单明细对应的第一条数据，将其放到状态中
                            lastJsonObjState.update(jsonObj);
                            //注册定时器
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        } else {
                            //如果状态中数据不为null，说明当前订单明细重复了，比较当前处理这条数据时间和状态中数据时间
                            String lastRowOpTs = lastJsonObj.getString("row_op_ts");
                            String curRowOpTs = jsonObj.getString("row_op_ts");
                            if (TimestampLtz3CompareUtil.compare(lastRowOpTs, curRowOpTs) <= 0) {
                                lastJsonObjState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        //定时器执行代码
                        JSONObject jsonObj = lastJsonObjState.value();
                        if (jsonObj != null) {
                            out.collect(jsonObj);
                        }
                        lastJsonObjState.clear();
                    }
                }
        );

        //TODO 7.将jsonObj转换为实体类对象
        SingleOutputStreamOperator<TradeCourseOrderBean> orderBeanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean map(JSONObject jsonObj) throws Exception {

                        String courseId = jsonObj.getString("course_id");
                        String userId = jsonObj.getString("user_id");
                        String orderId = jsonObj.getString("order_id");
                        Double finalAmount = jsonObj.getDouble("final_amount");
                        Long ts = jsonObj.getLong("ts") * 1000;

                        TradeCourseOrderBean orderBean = TradeCourseOrderBean.builder()
                                .courseId(courseId)
                                .userId(userId)
                                .orderIdSet(new HashSet(Collections.singleton(orderId)))
                                .finalAmount(finalAmount)
                                .orderUuCount(0L)
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );

//        orderBeanDS.print(">>>>");

        //TODO 8.按照用户id进行分组
        KeyedStream<TradeCourseOrderBean, String> userIdKeyedDS = orderBeanDS.keyBy(TradeCourseOrderBean::getUserId);

        //TODO 9.使用Flink状态编程判断是否是下单独立用户
        SingleOutputStreamOperator<TradeCourseOrderBean> uuOrderBeanDS = userIdKeyedDS.process(
                new KeyedProcessFunction<String, TradeCourseOrderBean, TradeCourseOrderBean>() {
                    private ValueState<String> lastOrderDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("lastOrderDateState", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastOrderDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(TradeCourseOrderBean orderBean, Context ctx, Collector<TradeCourseOrderBean> out) throws Exception {
                        //获取状态中的上次下单日期
                        String lastOrderDate = lastOrderDateState.value();
                        //获取今天的下单日期
                        String curOrderDate = DateFormatUtil.toDate(orderBean.getTs());

                        if (StringUtils.isEmpty(lastOrderDate) || !lastOrderDate.equals(curOrderDate)) {
                            orderBean.setOrderUuCount(1L);
                            lastOrderDateState.update(curOrderDate);
                        }
                        out.collect(orderBean);
                    }
                }
        );

        //TODO 10.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeCourseOrderBean> withWatermarkDS = uuOrderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeCourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeCourseOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeCourseOrderBean orderBean, long recordTimestamp) {
                                        return orderBean.getTs();
                                    }
                                }
                        )
        );

        //TODO 11.按照courseId维度进行分组
        KeyedStream<TradeCourseOrderBean, String> courseIdKeyedDS =
                withWatermarkDS.keyBy(TradeCourseOrderBean::getCourseId);

        //TODO 12.开窗
        WindowedStream<TradeCourseOrderBean, String, TimeWindow> windowDS = courseIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 13.聚合
        SingleOutputStreamOperator<TradeCourseOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean reduce(TradeCourseOrderBean value1, TradeCourseOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
                        value1.setFinalAmount(value1.getFinalAmount() + value2.getFinalAmount());
                        return value1;

                    }
                },
                new WindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeCourseOrderBean> input,
                                      Collector<TradeCourseOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());

                        for (TradeCourseOrderBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setTs(System.currentTimeMillis());
                            orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                            out.collect(orderBean);
                        }
                    }
                }
        );

//         reduceDS.print(">>>");

        //TODO 14.和Course维度进行关联
        SingleOutputStreamOperator<TradeCourseOrderBean> wihtSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeCourseOrderBean>("dim_course_info") {
                    @Override
                    public void join(TradeCourseOrderBean orderBean, JSONObject dimInfoJsonObj) {
                        orderBean.setCourseName(dimInfoJsonObj.getString("COURSE_NAME"));
                    }

                    @Override
                    public String getKey(TradeCourseOrderBean orderBean) {
                        return orderBean.getCourseId();
                    }
                },
                60, TimeUnit.SECONDS
        );

        //TODO 15.将聚合结果写到CK中
        wihtSkuInfoDS.print(">>>");
        wihtSkuInfoDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_trade_course_order_window values(?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
